package com.shujia.core

import com.mysql.jdbc.Driver
import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}

import java.sql.{Connection, DriverManager, ResultSet, Statement}
import scala.Range
import scala.collection.mutable.ListBuffer

object Code05Mysql2Text {
  def main(args: Array[String]): Unit = {

    val sc = new SparkContext(new SparkConf().setMaster("local").setAppName("Mysql2Text"))


    // 从Mysql中读取数据
    // 创建Mysql的连接对象




    //    for (elem <- Range(1, 1000)) {
    //      println(elem)
    //    }
    //    for (i <- 1 to 1000 ){
    //      println(i)
    //    }
    //    for (i <- 1 until 1000) {
    //      println(i)
    //    }

    //    val rangeRDD: RDD[Long] = sc.range(1500100001, 1500101001)
    //
    //    rangeRDD
    //      .map(
    //        selectId => {
    //  message from server: "Too many connections"
    //          DriverManager.registerDriver(new Driver())
    //          val connection: Connection = DriverManager.getConnection("jdbc:mysql://master:3306/stu?useSSL=false", "root", "123456")
    //          val statement: Statement = connection.createStatement()
    //          val set: ResultSet = statement.executeQuery(
    //            s"""
    //               |SELECT
    //               |*
    //               |FROM students
    //               |WHERE id= $selectId
    //               |""".stripMargin)
    //          var id: String = ""
    //          var clazz: String = ""
    //          while (set.next()) {
    //            id = set.getString(1)
    //            clazz = set.getString(5)
    //          }
    //          (id, clazz)
    //        }
    //      )
    //      .groupBy(_._2)
    //      .map {
    //        case (clazz, oneClazz) => (clazz, oneClazz.size)
    //      }
    //      .foreach(println)
    //

    sc
      .range(0, 3, 1, 3)
      .mapPartitions {
        case iterator => {
          val num: Long = iterator.toList(0)
          println("mapPartitions:"+num)
          List().toIterator
        }
      }.foreach(println)


    sc
      .range(0, 3, 1, 3)
      .mapPartitions {
        case iterator => {
          val num: Long = iterator.toList(0)
          println("mapPartitions执行了一次...")
          DriverManager.registerDriver(new Driver())
          val connection: Connection = DriverManager.getConnection("jdbc:mysql://master:3306/stu?useSSL=false", "root", "123456")
          val statement: Statement = connection.createStatement()
          val buffer: ListBuffer[(String, String)] = ListBuffer()
          for (elem <- Range(1500100001, 1500101001)) {
            if (elem % 3 == num) {
              val set: ResultSet = statement.executeQuery(
                s"""
                   |SELECT
                   |id
                   |,name
                   |,age
                   |,gender
                   |,substr(clazz,1,4)
                   |FROM students
                   |WHERE id= $elem
                   |""".stripMargin)

              while (set.next()) {
                buffer.append((set.getString(1), set.getString(5)))
              }
            }
          }

          buffer.iterator
        }
      }
      .groupBy(_._2)
      .map {
        case (clazz, oneClazz) => (clazz, oneClazz.size)
      }
      .foreach(println)

  }
}
